grpc::Status Subscribe{{ name.upper_camel_case }}(grpc::ServerContext* /* context */, const mavsdk::rpc::{{ plugin_name.lower_snake_case }}::Subscribe{{ name.upper_camel_case }}Request* {% if params %}request{% else %}/* request */{% endif %}, grpc::ServerWriter<rpc::{{ plugin_name.lower_snake_case }}::{{ name.upper_camel_case }}Response>* writer) override
{
    if (_lazy_plugin.maybe_plugin() == nullptr) {
        {% if has_result %}
            rpc::{{ plugin_name.lower_snake_case }}::{{ name.upper_camel_case }}Response rpc_response;
            {% if is_server %}
            // For server plugins, this should never happen, they should always be constructible.
            auto result = mavsdk::{{ plugin_name.upper_camel_case }}::Result::Unknown;
            {% else -%}
            auto result = mavsdk::{{ plugin_name.upper_camel_case }}::Result::NoSystem;
            {% endif -%}
            fillResponseWithResult(&rpc_response, result);
            writer->Write(rpc_response);
        {% endif %}
        return grpc::Status::OK;
    }

    auto stream_closed_promise = std::make_shared<std::promise<void>>();
    auto stream_closed_future = stream_closed_promise->get_future();
    register_stream_stop_promise(stream_closed_promise);

    auto is_finished = std::make_shared<bool>(false);
    auto subscribe_mutex = std::make_shared<std::mutex>();

    {% if not is_finite %}const mavsdk::{{ plugin_name.upper_camel_case }}::{{ name.upper_camel_case }}Handle handle = {% endif %}_lazy_plugin.maybe_plugin()->{% if not is_finite %}subscribe_{% endif %}{{ name.lower_snake_case }}{% if is_finite %}_async{% endif %}({% for param in params %}{% if not param.type_info.is_primitive %}translateFromRpc{{ param.name.upper_camel_case }}({% endif %}request->{{ param.name.lower_snake_case }}(){% if not param.type_info.is_primitive %}){% endif %}, {% endfor %}
        [this, &writer, &stream_closed_promise, is_finished, subscribe_mutex{% if not is_finite %}, &handle{% endif %}](
            {%- if has_result -%}mavsdk::{{ plugin_name.upper_camel_case }}::Result result,{%- endif -%}
            const {% if return_type.is_repeated %}std::vector<{% if not return_type.is_primitive %}{{ package.lower_snake_case.split('.')[0] }}::{{ plugin_name.upper_camel_case }}::{% endif %}{{ return_type.inner_name }}>{% else %}{%- if not return_type.is_primitive %}{{ package.lower_snake_case.split('.')[0] }}::{{ plugin_name.upper_camel_case }}::{% endif %}{{ return_type.name }}{% endif %} {{ name.lower_snake_case }}) {

        rpc::{{ plugin_name.lower_snake_case }}::{{ name.upper_camel_case }}Response rpc_response;
    {% if return_type.is_primitive %}
        rpc_response.set_{{ return_name.lower_snake_case }}({{ name.lower_snake_case }});
    {% elif return_type.is_enum %}
        rpc_response.set_{{ return_name.lower_snake_case }}(translateToRpc{{ return_type.name }}({{ name.lower_snake_case }}));
    {% elif return_type.is_repeated %}
        for (const auto& elem : {{ name.lower_snake_case }}) {
            auto* ptr = rpc_response.add_{{ return_name.lower_snake_case }}();
            ptr->CopyFrom(*translateToRpc{{ return_type.inner_name }}(elem).release());
        }
    {% else %}
        rpc_response.set_allocated_{{ return_name.lower_snake_case }}(translateToRpc{{ return_type.inner_name }}({{ name.lower_snake_case }}).release());
    {% endif %}

    {% if has_result %}
        auto rpc_result = translateToRpcResult(result);
        auto* rpc_{{ plugin_name.lower_snake_case }}_result = new rpc::{{ plugin_name.lower_snake_case }}::{{ plugin_name.upper_camel_case }}Result();
        rpc_{{ plugin_name.lower_snake_case }}_result->set_result(rpc_result);
        std::stringstream ss;
        ss << result;
        rpc_{{ plugin_name.lower_snake_case }}_result->set_result_str(ss.str());
        rpc_response.set_allocated_{{ plugin_name.lower_snake_case }}_result(rpc_{{ plugin_name.lower_snake_case }}_result);
    {% endif %}

        std::unique_lock<std::mutex> lock(*subscribe_mutex);
        if (!*is_finished && !writer->Write(rpc_response)) {
            {% if not is_finite %}
            _lazy_plugin.maybe_plugin()->unsubscribe_{{ name.lower_snake_case }}(handle);
            {% endif %}
            *is_finished = true;
            unregister_stream_stop_promise(stream_closed_promise);
            stream_closed_promise->set_value();
        }
    });

    stream_closed_future.wait();
    std::unique_lock<std::mutex> lock(*subscribe_mutex);
    *is_finished = true;

    return grpc::Status::OK;
}
